package com.clw.dwm;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.clw.KafkaUtils;
import com.clw.bean.OrderDetail;
import com.clw.bean.OrderInfo;
import com.clw.bean.OrderWide;
import com.clw.bean.Sku;
import com.clw.function.DimSinkFunction;
import com.clw.utils.DimUtil;
import com.clw.utils.JdbcUtil;
import com.clw.utils.MyKafkaUtil;

import com.clw.utils.PhonexUtil;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

//import org.springframework.boot.autoconfigure.SpringBootApplication;


/**
 * 处理宽表
 */
//@SpringBootApplication
public class


OrderWideApp {
    public static void main(String[] args) throws Exception {


        //⽅式⼀,本地和flink运⾏⽣效


        // 创建流环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 获取kafka 的流
        FlinkKafkaConsumer<String> consumer = MyKafkaUtil.getKafkaConsumer("mymaxwelltest", "assaas");
        //
        DataStreamSource<String> logStream = env.addSource(consumer);

        //  logStream.print();

        // 获取主表J
        SingleOutputStreamOperator<JSONObject> jsonSteam = logStream.map(x -> {
            return JSON.parseObject(x);
        });

        // 获取主表
        KeyedStream<OrderInfo, Long> orderInfoStream = jsonSteam.filter(x -> {
            return "order_info".equals(x.getString("table")) && "insert".equals(x.getString("type"));
        }).<OrderInfo>map(x -> {
            return JSON.toJavaObject(x.getJSONObject("data"), OrderInfo.class);
        }).keyBy(x -> {
            return x.getId();
        });


            /*
                .keyBy(x -> {
            return x.getId();
        });*/

        // 获取明细表的数据
        // 获取主表
        KeyedStream<OrderDetail, Long> orderDetailStream = jsonSteam.filter(x -> {
            return "order_detail".equals(x.getString("table")) && "insert".equals(x.getString("type"));
        }).map(x -> {
            return JSON.toJavaObject(x.getJSONObject("data"), OrderDetail.class);
        }).keyBy(x -> {
            return x.getOrder_id();
        });

        // 流的关联
        SingleOutputStreamOperator<OrderWide> wideStream = orderInfoStream.intervalJoin(orderDetailStream).between(Time.milliseconds(-2), Time.milliseconds(1)).process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {

            @Override
            public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>.Context context, Collector<OrderWide> collector) throws Exception {
                collector.collect(new OrderWide(orderInfo, orderDetail));
            }
        });

        wideStream.print();


        // 进行维度关联
        SingleOutputStreamOperator<OrderWide> processWdStream = wideStream.process(new ProcessFunction<OrderWide, OrderWide>() {
                                                                                       @Override
                                                                                       public void processElement(OrderWide order, ProcessFunction<OrderWide, OrderWide>.Context context, Collector<OrderWide> collector) throws Exception {

                                                                                           String sku_id = order.getSku_id().toString();
                                                                                           // 根据sku id 获取其他数据
                                                                                           JSONObject skuInfo = DimUtil.getDimInfo(JdbcUtil.getConnect(), "sku", "sku_id", sku_id, Sku.class);
                                                                                           order.setSku_name(skuInfo == null ? "没有找到sku名称" : skuInfo.getString("sku_name"));

                                                                                           collector.collect(order);
                                                                                       }
                                                                                   }

        );

        // 把流sink 到 hbase 的流程
        // 1. 检查表是否存在，如果不存在 则创建这个表
        PhonexUtil.checkTable("mywidetable1", OrderWide.class, "detail_id", "");

        // 2. 转换成工具类所需要的json 类型  注意  sinkTable  type  after
        processWdStream.map(x -> {
            // 将宽表的数据转换成json对象
            JSONObject jsonObject = new JSONObject();
            jsonObject.put("sinkTable", "mywidetable1");
            jsonObject.put("type", "insert");
            String jsonString = JSONObject.toJSONString(x);
            JSONObject afterJsonObj = JSON.parseObject(jsonString);
            jsonObject.put("after", afterJsonObj);
            return jsonObject;
        }).addSink(new DimSinkFunction() {
        });

        // 数据发送到kafka
        processWdStream.map(x -> {
            return JSON.toJSONString(x);
        }).addSink(KafkaUtils.createProduer("wide_table"));

        env.execute();


    }
}
